package com.xtck.ha.fault.core.service.impl;

import cn.hutool.core.date.DatePattern;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
import cn.hutool.http.Method;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.github.pagehelper.Page;
import com.xtck.ha.common.core.annotation.SortKey;
import com.xtck.ha.common.core.domain.R;
import com.xtck.ha.common.core.utils.PageUtils;
import com.xtck.ha.common.core.web.domain.AjaxResult;
import com.xtck.ha.common.core.web.service.dsimpl.BaseServiceImpl;
import com.xtck.ha.common.core.config.DsServerConfig;
import com.xtck.ha.common.core.utils.DomainUtils;
import com.xtck.ha.fault.core.model.DsFMEAInfo;
import com.xtck.ha.fault.core.model.DsFaultTask;
import com.xtck.ha.fault.core.model.DsModInfo;
import com.xtck.ha.fault.core.service.DsFMEAInfoService;
import com.xtck.ha.fault.core.service.DsFaultTaskService;
import com.xtck.ha.fault.core.service.DsModInfoService;
import com.xtck.ha.fault.framework.kafka.IKafkaProducer;
import com.xtck.ha.fault.framework.kafka.PushFaultMsg;
import com.xtck.ha.fault.framework.utils.JsonResult;
import com.xtck.ha.fault.framework.utils.MessageToJsonUtil;
import com.xtck.ha.fault.modules.pojo.entity.FaultTask;
import com.xtck.ha.fault.modules.pojo.entity.FmeaInfo;
import com.xtck.ha.fault.modules.pojo.entity.ModInfo;
import com.xtck.ha.fault.modules.pojo.vo.ResultDataVo;
import com.xtck.ha.fault.schedule.DynamicTaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.*;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;

import javax.annotation.Resource;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.stream.Collectors;


/**
 * @Description: DsFaultTaskService实现类
 * @Auther:liuyuan
 * @Date:2025-07-02
 */
@Slf4j
@Service
public class DsFaultTaskServiceImpl extends BaseServiceImpl<DsFaultTask> implements DsFaultTaskService {
    @Autowired
    DsModInfoService dsModInfoService;
    @Autowired
    DsFMEAInfoService dsFMEAInfoService;
    @Value("${web.predict-result}")
    private String predictResultUrl;
    @Autowired
    IKafkaProducer iKafkaProducer;
    @Resource
    private DynamicTaskService dynamicTaskService;
  /**
  *
  * 冗余代码，为自定义开发DS-Server接口服务
  *
  */
   private static String TABLE_NAME = "FAULT_TASK";
   private String getUrl(){
      return getDsServerConfig().getUrl();
   }

    @Override
    public List<DsFaultTask> selectDsFaultTaskList(DsFaultTask faultTask) {
        Page page = PageUtils.getLocalPage();

        RestTemplate restTemplate = new RestTemplate();
        Map<String, Object> map = DomainUtils.objectToMapForQuery(faultTask);
        JSONObject jo = new JSONObject();
        if (!map.isEmpty()){
            if (map.size()>1){
                List list = new ArrayList();
                map.forEach((k,v)->{
                    JSONObject pa = new JSONObject();
                    JSONObject like = new JSONObject();
                    if (k.equals("TASK_NAME")){
                        like.put("$like","%"+v+"%");
                        pa.put(k,like);
                    }else
                        pa.put(k,v);
                    list.add(pa);
                });
                jo.put("$and",list);
            }else {
                jo.putAll(map);
            }
        }
        UriComponentsBuilder ucb = UriComponentsBuilder.fromHttpUrl(geturl());
        if(!map.isEmpty()){
            ucb = ucb.queryParam("filter", jo.toJSONString());
        }
        if (page!=null){
            ucb = ucb.queryParam("page",page.getPageNum())
                    .queryParam("pagesize",page.getPageSize());
        }

        // 新增排序
        if (entityClass.isAnnotationPresent(SortKey.class)){
            String value = this.entityClass.getAnnotation(SortKey.class).value();
            ucb = ucb.queryParam("sort", value);
        }

        URI uri = ucb.build().toUri();
        // 设置请求头，特别是Authorization头部来携带token
        HttpHeaders headers = new HttpHeaders();
        headers.add("Ds-Route-key", this.getDsRouteKey());
        // 不需要传递body对于GET请求来说，除非你需要传递JSON等格式的参数体，但通常不这么做。
        HttpEntity<String> entity = new HttpEntity<>(headers);
        // 执行GET请求并处理响应
        R response = restTemplate.exchange(uri, HttpMethod.GET, entity, R.class).getBody();
//        log.info("Response with params: {}",response.getData());
        PageUtils.clearPage();
        List<DsFaultTask> list = new ArrayList<>();
        if (R.isSuccess(response)) {
            Object data1 = response.getData();
            if (data1 != null) {
                JSONArray arrays = com.alibaba.fastjson2.JSON.parseArray(JSON.toJSONString(data1));
                list = arrays.stream().map(o -> {
                    try {
                        return JSONObject.parseObject(JSONObject.toJSONString(o), this.getEntityClass());
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }).collect(Collectors.toList());
            }
        }
        return list;
    }

    @Override
    public long countDsFaultTask(DsFaultTask faultTask) {
        RestTemplate restTemplate = new RestTemplate();
        Map<String, Object> map = DomainUtils.objectToMapForQuery(faultTask);
        JSONObject jo = new JSONObject();
        if (!map.isEmpty()){
            if (map.size()>1){
                List list = new ArrayList();
                map.forEach((k,v)->{
                    JSONObject pa = new JSONObject();
                    JSONObject like = new JSONObject();
                    if (k.equals("TASK_NAME")){
                        like.put("$like","%"+v+"%");
                    }
                    pa.put(k,like);
                    list.add(pa);
                });
                jo.put("$and",list);
            }else {
                jo.putAll(map);
            }
        }
        UriComponentsBuilder ucb = UriComponentsBuilder.fromHttpUrl(geturl()+"/_size");
        if(!map.isEmpty()){
            ucb = ucb.queryParam("filter", jo.toJSONString());
        }
        URI uri = ucb.build().toUri();
        // 设置请求头，特别是Authorization头部来携带token
        HttpHeaders headers = new HttpHeaders();
        headers.add("Ds-Route-key", this.getDsRouteKey());
        // 不需要传递body对于GET请求来说，除非你需要传递JSON等格式的参数体，但通常不这么做。
        HttpEntity<String> entity = new HttpEntity<>(headers);
        // 执行GET请求并处理响应
        ResponseEntity<R> response = restTemplate.exchange(uri, HttpMethod.GET, entity, R.class);
        log.info("Response with params: {}",response.getBody().getData());
        if (R.isSuccess(response.getBody())) {
            JSONObject data = JSONObject.from(response.getBody().getData());
            return data.getLongValue("_size");
        }
        return 0;
    }

    /**只执行一次计算，不使用定时任务流程，不考虑下次计算*/
    @Override
    @Transactional(rollbackFor = Exception.class)
    public AjaxResult computeOnce(DsFaultTask faultTask){
        /**查询对应模型信息*/
        List<DsModInfo> modInfoList = dsModInfoService.selectTList(DsModInfo.builder().modName(faultTask.getModelName()).build());
        if (null == modInfoList ){
            return AjaxResult.error("未查询到" +faultTask.getModelName()+"模型");
        }
        else if (modInfoList.size() != 1){
            return AjaxResult.error("模型名称重复");
        }
        else {
            /**构建参数，调用查询数据接口*/
            DsModInfo modInfo = modInfoList.get(0);
            ResultDataVo resultDataVo = ResultDataVo.builder()
                    .satelliteCode(faultTask.getSatelliteCode())
                    .paramCode(faultTask.getParamCode())
                    .startTime(faultTask.getStartTime())
                    .endTime(faultTask.getEndTime())
                    .build();
            HttpResponse getPredictDataRequest = null;
            try {
                getPredictDataRequest = HttpUtil.createRequest(Method.POST, predictResultUrl).body(com.alibaba.fastjson.JSON.toJSONString(resultDataVo)).execute();
            } catch (Exception e) {
                return AjaxResult.error("调用接口失败：" + e.getMessage());
            }
            if (null != getPredictDataRequest && getPredictDataRequest.getStatus() == 200){
                com.alibaba.fastjson.JSONObject result = com.alibaba.fastjson.JSON.parseObject(getPredictDataRequest.body());
                com.alibaba.fastjson.JSONArray dataList = result.getJSONArray("data");
                /**查询不到数据就停止任务*/
                if (null==dataList || dataList.size()==0){
                    return AjaxResult.error("未查询到" +faultTask.getParamCode()+ "数据");
                }else {
                    Map inputParam =new HashMap<String,Object>();
                    inputParam.put("dataList",dataList);
                    String getFaultInfoRequestBody = com.alibaba.fastjson.JSON.toJSONString(inputParam);
                    /**调用故障模型计算接口*/
                    HttpResponse getFaultInfoRequest = null;
                    try {
                        getFaultInfoRequest = HttpUtil.createRequest(Method.POST, modInfo.getAddr()).body(getFaultInfoRequestBody).execute();
                    } catch (Exception e) {
                        return AjaxResult.error("调用接口失败：" + e.getMessage());
                    }
                    if (null != getFaultInfoRequest && getFaultInfoRequest.getStatus() == 200){
                        /**获取返回的故障信息并入库*/
                        com.alibaba.fastjson.JSONObject faultInfoResult = com.alibaba.fastjson.JSON.parseObject(getFaultInfoRequest.body());
                        com.alibaba.fastjson.JSONObject faultInfo = faultInfoResult.getJSONObject("data");
                        DsFMEAInfo fmeaInfo = com.alibaba.fastjson.JSON.toJavaObject(faultInfo, DsFMEAInfo.class);
                        fmeaInfo.setCreateTime(new Date());
                        fmeaInfo.setStartTime(resultDataVo.getStartTime());
                        fmeaInfo.setEndTime(resultDataVo.getEndTime());
                        fmeaInfo.setTaskId(faultTask.getId());
                        dsFMEAInfoService.insertT(fmeaInfo);

                        /**构建消息体并发送kafka消息*/
                        SimpleDateFormat sdf = new SimpleDateFormat(DatePattern.NORM_DATETIME_PATTERN);
                        PushFaultMsg pushFaultMsg = PushFaultMsg.builder()
                                .sendTime(sdf.format(new Date()))
                                .failureMode(fmeaInfo.getFailureMode())
                                .effect(fmeaInfo.getEffect())
                                .severity(fmeaInfo.getSeverity())
                                .cause(fmeaInfo.getCause())
                                .occurrence(fmeaInfo.getOccurrence())
                                .currentProcessControlPlan(fmeaInfo.getCurrentProcessControlPlan())
                                .detection(fmeaInfo.getDetection())
                                .rpn(fmeaInfo.getRpn())
                                .recommendedAction(fmeaInfo.getRecommendedAction())
                                .build();
                        iKafkaProducer.send("FAULT_RESULT",pushFaultMsg);
                        log.info("发送的kafka信息为============="+ MessageToJsonUtil.toJson(pushFaultMsg));

                        return AjaxResult.success("调用" +faultTask.getModelName()+ "模型计算接口成功");
                    }
                    else {
                        return AjaxResult.error("调用" +faultTask.getModelName()+ "模型计算接口失败");
                    }
                }
            }
            else{
                return AjaxResult.error("查询" +faultTask.getParamCode()+ "数据接口调用失败");
            }
        }
    }

    @Override
    public int save(DsFaultTask faultTask){
        faultTask.setIsFirstExec(true);
        faultTask.setCreateTime(new Date());
        faultTask.setStatus(true);
        return this.insertT(faultTask);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public AjaxResult execTask(DsFaultTask faultTask) {
        String cron = checkCron(faultTask.getCron());
        //  定义任务逻辑
        Runnable task = () -> {
            faultProcess(faultTask);
        };
        //  启动任务
        dynamicTaskService.startOrUpdateTask(String.valueOf(faultTask.getId()), cron, task);
        DsFaultTask taskForDb = this.selectTById(faultTask.getId());
        if (null != taskForDb && !taskForDb.getStatus()){
            taskForDb.setStatus(true);
            this.updateT(taskForDb);
        }
        return AjaxResult.success("任务启动成功: " + faultTask.getTaskName());
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public AjaxResult stopTask(DsFaultTask faultTask) {
        dynamicTaskService.stopTask(String.valueOf(faultTask.getId()));
        DsFaultTask taskForDb = this.selectTById(faultTask.getId());
        if (null != taskForDb && taskForDb.getStatus()){
            taskForDb.setStatus(false);
            this.updateT(taskForDb);
            return AjaxResult.success("任务停止成功: " + taskForDb.getTaskName());
        }
        else
            return AjaxResult.error("任务停止失败: " + taskForDb.getTaskName());
    }

    /**检查cron表达式是不是6位*/
    private String checkCron(String cron){
        String newCron = "";
        String[] numstr = cron.split("\\s+");
        if (numstr.length > 6){
            for (int i=0; i<5 ;i++){
                newCron += numstr[i] + " ";
            }
            newCron += numstr[5];
            return newCron;
        }
        else
            return cron;
    }

    private void faultProcess(DsFaultTask faultTask){
        /**查询对应模型信息*/
        List<DsModInfo> modInfoList = dsModInfoService.selectTList(DsModInfo.builder().modName(faultTask.getModelName()).build());
        if (null == modInfoList ){
            log.error("未查询到对应模型");
            return;
        }
        else if (modInfoList.size() != 1){
            log.error("模型名称重复");
            return;
        }
        else {
            /**构建参数，调用查询数据接口*/
            DsModInfo modInfo = modInfoList.get(0);
            ResultDataVo resultDataVo = ResultDataVo.builder()
                    .satelliteCode(faultTask.getSatelliteCode())
                    .paramCode(faultTask.getParamCode())
                    .build();
            /**是否是第一次执行将会影响时间段参数*/
            if (faultTask.getIsFirstExec()){
                resultDataVo.setStartTime(faultTask.getStartTime());
                resultDataVo.setEndTime(faultTask.getEndTime());
            }
            else {
                long startTime = faultTask.getStartTime().getTime();
                long endTime = faultTask.getEndTime().getTime();
                Date newStartTime = faultTask.getEndTime();
                Date newEndTime = new Date(endTime + (endTime-startTime));
                resultDataVo.setStartTime(newStartTime);
                resultDataVo.setEndTime(newEndTime);
            }
            HttpResponse getPredictDataRequest = null;
            try {
                getPredictDataRequest = HttpUtil.createRequest(Method.POST, predictResultUrl).body(com.alibaba.fastjson.JSON.toJSONString(resultDataVo)).execute();
            } catch (Exception e) {
                log.error("调用接口失败：" + e.getMessage());
                return;
            }
            if (null != getPredictDataRequest && getPredictDataRequest.getStatus() == 200){
                com.alibaba.fastjson.JSONObject result = com.alibaba.fastjson.JSON.parseObject(getPredictDataRequest.body());
                com.alibaba.fastjson.JSONArray dataList = result.getJSONArray("data");
                /**查询不到数据就停止任务*/
                if (null==dataList || dataList.size()==0){
                    log.error("未查询到对应数据");
                    return;
                }else {
                    Map inputParam =new HashMap<String,Object>();
                    inputParam.put("dataList",dataList);
                    String getFaultInfoRequestBody = com.alibaba.fastjson.JSON.toJSONString(inputParam);
                    /**调用故障模型计算接口*/
                    HttpResponse getFaultInfoRequest = null;
                    try {
                        getFaultInfoRequest = HttpUtil.createRequest(Method.POST, modInfo.getAddr()).body(getFaultInfoRequestBody).execute();
                    } catch (Exception e) {
                        log.error("调用接口失败：" + e.getMessage());
                        return;
                    }
                    if (null != getFaultInfoRequest && getFaultInfoRequest.getStatus() == 200){
                        /**获取返回的故障信息并入库*/
                        com.alibaba.fastjson.JSONObject faultInfoResult = com.alibaba.fastjson.JSON.parseObject(getFaultInfoRequest.body());
                        com.alibaba.fastjson.JSONObject faultInfo = faultInfoResult.getJSONObject("data");
                        DsFMEAInfo fmeaInfo = com.alibaba.fastjson.JSON.toJavaObject(faultInfo, DsFMEAInfo.class);
                        fmeaInfo.setCreateTime(new Date());
                        fmeaInfo.setStartTime(resultDataVo.getStartTime());
                        fmeaInfo.setEndTime(resultDataVo.getEndTime());
                        fmeaInfo.setTaskId(faultTask.getId());
                        dsFMEAInfoService.insertT(fmeaInfo);

                        /**构建消息体并发送kafka消息*/
                        SimpleDateFormat sdf = new SimpleDateFormat(DatePattern.NORM_DATETIME_PATTERN);
                        PushFaultMsg pushFaultMsg = PushFaultMsg.builder()
                                .sendTime(sdf.format(new Date()))
                                .failureMode(fmeaInfo.getFailureMode())
                                .effect(fmeaInfo.getEffect())
                                .severity(fmeaInfo.getSeverity())
                                .cause(fmeaInfo.getCause())
                                .occurrence(fmeaInfo.getOccurrence())
                                .currentProcessControlPlan(fmeaInfo.getCurrentProcessControlPlan())
                                .detection(fmeaInfo.getDetection())
                                .rpn(fmeaInfo.getRpn())
                                .recommendedAction(fmeaInfo.getRecommendedAction())
                                .build();
                        iKafkaProducer.send("FAULT_RESULT",pushFaultMsg);
                        log.info("发送的kafka信息为============="+ MessageToJsonUtil.toJson(pushFaultMsg));

                        /**更新任务信息，查询数据时间段变化*/
                        faultTask.setStartTime(resultDataVo.getStartTime());
                        faultTask.setEndTime(resultDataVo.getEndTime());
                        faultTask.setIsFirstExec(false);
                        this.updateT(faultTask);

                    }
                    else {
                        log.error("调用模型计算接口失败");
                        return;
                    }
                }

            }
            else{
                log.error("查询数据失败");
                return;
            }
        }
    }
}
